use std::time::{Duration, SystemTime, UNIX_EPOCH};
use std::sync::{Mutex, Arc};
use std::sync::mpsc::{Receiver, SyncSender};
use std::collections::HashMap;
use global_counter::primitive::exact::CounterU64;
use std::thread::sleep;
use tokio::runtime::Handle;

type LockerMap = HashMap<String, Item>;
type Item = Arc<Mutex<LockItem>>;

const DEFAULT_WAIT_TIME: u64 = 1;
const DEFAULT_DURATION: u64 = 3;

#[derive(PartialEq, Eq, Clone, Copy)]
enum State {
    // 未加锁
    Free,
    // 被他人锁住
    LockedByOther,
    // 被我锁住
    LockedByMe,
    // 死锁
    Dead,
}

#[derive(PartialEq, Eq, Clone, Copy)]
pub enum LockResult {
    // 成功
    _OK,
    // 获取锁超时
    WaitTimeout,
    // 已经死锁
    DeadLock,
}

pub struct LockItem {
    // 锁计数
    counter: i32,

    // 被他人解开通知
    sig_sender: SyncSender<bool>,
    sig_receiver: Arc<Mutex<Receiver<bool>>>,

    // 锁的拥有者
    owner: u64,

    // 序列号
    serial: u64,
}

lazy_static! {
    static ref ID_INC:CounterU64 = CounterU64::new(0);
    static ref OWNER_MAP:Arc<Mutex<HashMap<u64, LockerMap >>> = Arc::new(Mutex::new(HashMap::new()));
    static ref KEY_MAP:Arc<Mutex<LockerMap >> = Arc::new(Mutex::new(HashMap::new()));
    static ref BLOCKING_MAP:Arc<Mutex<HashMap<u64, Item>>> = Arc::new(Mutex::new(HashMap::new()));
    static ref HANDLE: Mutex<Option<Handle>> = Mutex::new(None);
}

// init 初始化
pub fn init() {
    *HANDLE.lock().unwrap() = Some(Handle::current());
}

// lock 请求加锁
// key 锁的key
// requester 请求者
// Result<序列号, 错误>
pub fn lock(key: &String, requetser: u64) -> Result<u64, LockResult> {
    println!("{} start lock {}", requetser, key);
    // 确保序列号唯一
    let serial = gen_record_id();
    // 等待锁释放
    let r = wait(key, requetser);
    match r {
        State::Free => {}
        State::LockedByOther => {
            return Err(LockResult::WaitTimeout);
        }
        State::LockedByMe => {}
        State::Dead => { // 检测到死锁
            return Err(LockResult::DeadLock);
        }
    }

    let item = make_or_get_item(key, requetser, serial);
    let mut it_clone = item.lock().unwrap();
    it_clone.counter += 1;

    let kc = key.clone();
    let handle = HANDLE.lock().unwrap().as_ref().unwrap().clone();
    // 开始等待锁过期
    handle.spawn(async move {
        sleep(Duration::from_secs(DEFAULT_DURATION));
        println!("{} expired", kc);
        unlock(&kc, requetser, serial);
    });

    println!("{} locked by {}", key, requetser);
    return Ok(serial);
}

// unlock 请求解锁
// key 锁的key
// owner 锁的拥有者
// serial 序列号
pub fn unlock(key: &String, owner: u64, serial: u64) {
    // 从key表删除
    let mut m = KEY_MAP.lock().unwrap();
    let o = m.remove(key);
    if o.is_none() {
        return;
    }
    let item = o.unwrap();
    let mut li = item.lock().unwrap();
    if li.owner != owner || li.serial != serial {
        return;
    }
    li.counter -= 1;
    if li.counter > 0 {
        return;
    }
    // 从拥有者表删除
    let mut om = OWNER_MAP.lock().unwrap();
    let imo = om.get_mut(&owner);
    match imo {
        None => {}
        Some(im) => {
            im.remove(key);
            if im.is_empty() {
                om.remove(&owner);
            }
        }
    }
    // 通知等待者，锁已经释放
    let _ = li.sig_sender.send(true);
    println!("{} unlocked", key)
}

// -----------------------------------------------------------------------------------------

fn gen_record_id() -> u64 {
    let mi = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis();
    (mi as u64) << 32 | ID_INC.inc()
}

fn make_item(owner: u64, serial: u64) -> Item {
    let release_sig = std::sync::mpsc::sync_channel(1);
    let new_item = LockItem {
        counter: 0,
        sig_sender: release_sig.0.clone(),
        sig_receiver: Arc::new(Mutex::new(release_sig.1)),
        owner,
        serial,
    };
    return Arc::new(Mutex::new(new_item));
}

fn make_or_get_item(key: &String, owner: u64, serial: u64) -> Item {
    let mut m = KEY_MAP.lock().unwrap();
    return if m.contains_key(key) {
        m.get(key).unwrap().clone()
    } else {
        let new_item = make_item(owner, serial);
        m.insert(key.clone(), new_item.clone());
        let mut om = OWNER_MAP.lock().unwrap();
        let olm = om.entry(owner).or_insert(LockerMap::new());
        olm.entry(key.clone()).or_insert(new_item.clone());
        new_item.clone()
    };
}

fn add_blocking(owner: u64, item: Item) {
    let mut bm = BLOCKING_MAP.lock().unwrap();
    bm.insert(owner, item);
}

fn remove_blocking(owner: u64) {
    let mut bm = BLOCKING_MAP.lock().unwrap();
    bm.remove(&owner);
}

fn dead_detect(requester: u64, item: Item) -> bool {
    let bm_clone = BLOCKING_MAP.clone();
    let bm = bm_clone.lock().unwrap();
    let li = item.lock().unwrap();
    let mut o = bm.get(&li.owner);
    drop(li);
    for _ in 0..9 {
        match o {
            None => {
                return false;
            }
            Some(item) => {
                let li = item.lock().unwrap();
                if li.owner == requester {
                    return true;
                }
                o = bm.get(&li.owner);
            }
        }
    }
    return false;
}

fn wait(key: &String, requester: u64) -> State {
    let m = KEY_MAP.lock().unwrap();
    let o = m.get(key);
    if o.is_none() {
        return State::Free;
    }

    let mut item = o.unwrap().lock().unwrap();
    if item.owner == requester {
        return State::LockedByMe;
    }
    drop(item);
    if dead_detect(requester, o.unwrap().clone()) {
        println!("{} lock {} detected dead", requester, key);
        return State::Dead;
    }
    item = o.unwrap().lock().unwrap();
    add_blocking(requester, o.unwrap().clone());
    let recv_clone = item.sig_receiver.clone();
    let receiver = recv_clone.lock().unwrap();
    let sender = item.sig_sender.clone();
    drop(item);
    drop(m);

    let r = receiver.recv_timeout(Duration::new(DEFAULT_WAIT_TIME, 0));
    return match r {
        Ok(_) => {
            remove_blocking(requester);
            let _ = sender.send(true);
            State::Free
        }
        Err(_) => {
            println!("{} wait {} timeout", requester, key);
            remove_blocking(requester);
            State::LockedByOther
        }
    };
}
